package com.fei.kafka;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.alibaba.fastjson.JSONObject;

public class KafkaProducerTest {

	private void execMsgSend() throws InterruptedException {
		Properties props = new Properties();
		props.put("zookeeper.connect", "node1:2181,node1:2182,node1:2183/cluster1");// 声明zk
		props.put("bootstrap.servers", "node2:9092,node2:9093,node2:9091");// 声明kafka
																			// broker
		props.put("group.id", "testGroup");
		props.put("batch.size", "16384");
		props.put("retries", "10");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("request.required.acks", "0");

		KafkaProducer<String, String> procuder = new KafkaProducer<>(props);

		/*
		 * String topic = "access"; new SendMsgThread(procuder, topic).start();
		 * new SendMsgThread(procuder, "serverinfo").start();
		 */
		try {
			String topic = "access";
			String data = "{\"fid\":\"currentOrder\",\"app\":\"blancesoa\",\"rule_status\":\"1\",\"host\":\"ordernew.m.jd.local\",\"ts\":\"1492675216\",\"client\":\"android\",\"location\":\"\",\"xhLevel\":\"\",\"ver\":\"6.0.0\",\"freqLevel\":\"1\",\"freqSw\":\"1\",\"ip\":\"10.12.181.17\",\"isxh\":\"0\",\"sip\":\"10.189.122.81\",\"forcebot\":\"\",\"pin\":\"lizexpertise\",\"umg\":\"50\",\"rule_level\":\"2\",\"isDeny\":\"0\",\"uuid\":\"jd228233vbu\",\"isIntercept\":false,\"sw\":\"0\",\"ts\":1495435892000}";
			JSONObject json = JSONObject.parseObject(data);
			while (true) {
				long ts = System.currentTimeMillis();
				json.put("ts", ts);
				String value = json.toJSONString();
				ProducerRecord<String, String> msg = new ProducerRecord<String, String>(topic, value);
				procuder.send(msg);

				TimeUnit.SECONDS.sleep(1);
			}
		} finally {
			procuder.close();
		}

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		KafkaProducerTest simpleProducer = new KafkaProducerTest();
		simpleProducer.execMsgSend();
	}
}

class SendMsgThread extends Thread {
	private KafkaProducer<String, String> procuder;
	private String topic;

	public SendMsgThread(KafkaProducer<String, String> procuder, String topic) {
		this.procuder = procuder;
		this.topic = topic;
	}

	@Override
	public void run() {
		String data = "{\"fid\":\"currentOrder\",\"app\":\"blancesoa\",\"rule_status\":\"1\",\"host\":\"ordernew.m.jd.local\",\"ts\":\"1492675216\",\"client\":\"android\",\"location\":\"\",\"xhLevel\":\"\",\"ver\":\"6.0.0\",\"freqLevel\":\"1\",\"freqSw\":\"1\",\"ip\":\"10.12.181.17\",\"isxh\":\"0\",\"sip\":\"10.189.122.81\",\"forcebot\":\"\",\"pin\":\"lizexpertise\",\"umg\":\"50\",\"rule_level\":\"2\",\"isDeny\":\"0\",\"uuid\":\"jd228233vbu\",\"isIntercept\":false,\"sw\":\"0\",\"ts\":1495435892000}";
		JSONObject json = JSONObject.parseObject(data);
		while (true) {
			for (int i = 1; i <= 1000; i++) {
				long ts = System.currentTimeMillis();
				json.put("ts", ts);
				String value = json.toJSONString();
				ProducerRecord<String, String> msg = new ProducerRecord<String, String>(topic, value);
				procuder.send(msg);
			}

			try {
//				TimeUnit.SECONDS.sleep(1);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}